package com.shujia.flink.source;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

public class Demo2FileSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //执行模式
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        /*
         * file source -- 有界流
         */
        //老版本api
        DataStreamSource<String> fileDS = env.readTextFile("flink/data/words.txt");

        //fileDS.print();


        /**
         * flink新版的file source
         * flink 可以基于目录进行流处理
         *
         * flink流批统一
         * 1、flink基于文件可以做批处理，也可以做流处理
         */

        //创建file source
        FileSource<String> fileSource = FileSource
                .forRecordStreamFormat(
                        new TextLineInputFormat(),//指定读取数据的格式
                        new Path("flink/data/student")//指定读取数据䣌路径
                )
                //切换成无界流，每隔一段时间扫描目录下新的文件
                //.monitorContinuously(Duration.ofSeconds(5))
                .build();

        //使用file source
        DataStream<String> studentDS = env
                .fromSource(fileSource, WatermarkStrategy.noWatermarks(), "fileSource");

        DataStream<Tuple2<String, Integer>> kvDS = studentDS.map(line -> {
            String[] split = line.split(",");
            String clazz = split[4];
            return Tuple2.of(clazz, 1);
        }, Types.TUPLE(Types.STRING, Types.INT));

        //统计班级的人数
        DataStream<Tuple2<String, Integer>> clazzNumDS = kvDS
                .keyBy(kv -> kv.f0)
                .sum(1);

        clazzNumDS.print();

        env.execute();

    }
}
